package org.baojie.raft.node;

import org.baojie.raft.box.Box;

import org.baojie.raft.message.HeartBeatPing;
import org.baojie.raft.message.MsgID;
import org.baojie.raft.message.RaftMessage;
import org.baojie.raft.message.RaftMsgSignal;
import org.baojie.raft.node.type.Follower;
import org.baojie.raft.node.type.Role;
import org.baojie.raft.store.NodeStore;
import org.baojie.raft.util.current.LocalSleep;
import org.baojie.raft.util.current.RaftFactory;
import org.baojie.raft.util.current.RaftThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicStampedReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

// raft 内核做的仅仅是 自动机的定时器的周期运动
public abstract class AbstractNode implements CoreInNode {

    private static final Logger log = LoggerFactory.getLogger(AbstractNode.class);

    // 节点自身的定时自动机
    private final RaftThreadPool autoHeartPool = new RaftThreadPool(
            4,
            32,
            60,
            TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            RaftFactory.create("auto wait heart beat in node", false));

    private final RaftThreadPool noHeartPool = new RaftThreadPool(
            4,
            32,
            60,
            TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            RaftFactory.create("call box no heart work in node", false));

    private final RaftThreadPool dealHeartPool = new RaftThreadPool(
            4,
            32,
            60,
            TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            RaftFactory.create("call box deal heart work in node", false));

    private final RaftThreadPool autoSendHeartPool = new RaftThreadPool(
            4,
            32,
            60,
            TimeUnit.SECONDS,
            new SynchronousQueue<>(),
            RaftFactory.create("auto send heart work in node", false));

    // 节点自身的随机数发生器
    private final ThreadLocalRandom random = ThreadLocalRandom.current();
    // 节点的开关状态
    private final AtomicBoolean start = new AtomicBoolean(false);

    // 节点接受心跳信息的同步信号
    private final SynchronousQueue<RaftMessage> heartNotify = new SynchronousQueue<>();

    private final List<Future<?>> futures = new ArrayList<>(128);

    // 节点状态控制的同步锁
    private final ReentrantLock mainLock = new ReentrantLock();

    // 当前节点的 名称
    private final NodeName name;
    // 节点当前在raft集群中的角色
    private final AtomicStampedReference<Role> role;
    private final NodeStore store;
    private final Box box;

    public AbstractNode(Box box) {
        this.box = box;
        this.store = box.store();
        NodeName voteFor = store.getVotedFor();
        int term = store.getTerm();
        this.role = new AtomicStampedReference<>(new Follower(voteFor, null, store, term), 0);
        this.name = box.name();
    }

    public boolean touch() {
        final Lock lock = mainLock;
        return lock.tryLock();
    }

    public void unTouch() {
        final Lock lock = mainLock;
        lock.unlock();
    }

    @Override
    public Box box() {
        return box;
    }

    @Override
    public NodeName name() {
        return name;
    }

    @Override
    public Role role() {
        return role.getReference();
    }

    public int roleStamp() {
        return role.getStamp();
    }

    public boolean stampSetRole(Role older, Role newer, int oldStamp) {
        return role.compareAndSet(older, newer, oldStamp, oldStamp + 1);
    }


    // 这个地方需要加锁
    @Override
    public void start() {
        if (start.get()) {
            return;
        } else {
            if (start.compareAndSet(false, true)) {
                doStart();
            }
        }
    }

    private void doStart() {
        WaitHeartBeatWorker heart = new WaitHeartBeatWorker();
        futures.add(autoHeartPool.submit(heart));
        SendHeartTask send = new SendHeartTask();
        futures.add(autoSendHeartPool.submit(send));
    }

    // 这个地方需要加锁
    @Override
    public void stop() {
        if (start.get()) {
            if (start.compareAndSet(true, false)) {
                stopByLock();
            }
        } else {
            return;
        }
    }

    private void stopByLock() {
        for (Future<?> fu : futures) {
            fu.cancel(true);
        }
        autoHeartPool.shutdown(autoHeartPool);
        dealHeartPool.shutdown(dealHeartPool);
        noHeartPool.shutdown(noHeartPool);
        autoSendHeartPool.shutdown(autoSendHeartPool);
    }

    public boolean putPing(RaftMessage ping) {
        try {
            return heartNotify.offer(ping, 5600, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            return false;
        }
    }

    private final class WaitHeartBeatWorker implements Runnable {

        private final AtomicBoolean exist = new AtomicBoolean(false);

        private long lastTime = System.currentTimeMillis();
        private String perMsgId = null;

        public WaitHeartBeatWorker() {

        }

        @Override
        public void run() {
            LocalSleep.park(voteRandomMillis(), TimeUnit.MILLISECONDS);
            for (; ; ) {
                if (worldStopped()) {
                    return;
                } else {
                    RaftMessage ping = waitHeartBeat();
                    Role role = role();
                    // 如果是leader状态
                    // 心跳超时不需要考虑
                    if (role.type() == Role.Name.LEADER) {
                        LocalSleep.safePoint();
                        //log.info("auto wait heart beat, i am leader , name = " + name);
                        System.out.println("auto wait heart beat, i am leader , name = " + name);
                        continue;
                    }
                    // 一定是获取到锁之后根据状态判断
                    // 不能够tryLock
                    final Lock lock = mainLock;
                    lock.lock();
                    try {
                        long now = System.currentTimeMillis();
                        role = role();
                        int stamp = roleStamp();
                        if (role.type() == Role.Name.LEADER) {
                            LocalSleep.safePoint();
                            //log.info("auto wait heart beat, i am leader , name = " + name);
                            System.out.println("auto wait heart beat, i am leader , name = " + name);
                            lastTime = now;
                            continue;
                        }
                        // 如果被取消了需要立即停止
                        if (worldStopped()) {
                            return;
                        }
                        // 是一个自动周期运行的流程
                        // 所以这里仅仅判断状态即可不需要加锁

                        if ((now - lastTime) < Meta.min_elc_time.millis()) {
                            LocalSleep.safePoint();
                            lastTime = now;
                            continue;
                        }

                        if (null == ping) {
                            // 没有获取到心跳信号
                            if (null == perMsgId) {
                                perMsgId = MsgID.getInstance().msgId();
                                RaftMsgSignal tmp = box.touchStartVote(perMsgId);
                                if (null == tmp) {
                                    noHeartBeat(role, stamp, perMsgId);
                                } else {
                                    log.error("存在心跳线程之外的线程操作缓存的放入动作");
                                }
                            } else {
                                RaftMsgSignal tmp = box.touchStartVote(perMsgId);
                                if (null == tmp) {
                                    perMsgId = MsgID.getInstance().msgId();
                                    noHeartBeat(role, stamp, perMsgId);
                                } else {
                                    log.warn("等待选举结果的线程还未结束");
                                }
                            }
                        } else {
                            // 存在处理心跳信号的线程
                            if (exist.get()) {
                                log.warn("存在处理心跳信号的线程");
                                continue;
                            }
                            // 采取异步模式处理心跳
                            if (exist.compareAndSet(false, true)) {
                                DealHeartPing dealPing = new DealHeartPing(exist, ping, role, stamp);
                                dealHeartPool.submit(dealPing);
                            }
                        }
                    } finally {
                        lock.unlock();
                    }
                }
            }
        }

    }

    private void noHeartBeat(Role oldRole, int stamp, String msgId) {
        // box 负责处理 follower 与 candidate 情况
        // 异步模式处理
        NoHeartPing noPing = new NoHeartPing(oldRole, stamp, msgId);
        noHeartPool.submit(noPing);
    }

    private final class NoHeartPing implements Runnable {

        private final Role oldRole;
        private final int stamp;
        private final String msgId;

        public NoHeartPing(Role oldRole, int stamp, String msgId) {
            this.oldRole = oldRole;
            this.stamp = stamp;
            this.msgId = msgId;
        }

        @Override
        public void run() {
            box.doNodeVote(oldRole, stamp, msgId);
        }

    }

    private final class DealHeartPing implements Runnable {
        private final AtomicBoolean doing;
        private final RaftMessage ping;
        private final Role oldRole;
        private final int oldStamp;


        public DealHeartPing(AtomicBoolean doing, RaftMessage ping, Role oldRole, int oldStamp) {
            this.doing = doing;
            this.ping = ping;
            this.oldRole = oldRole;
            this.oldStamp = oldStamp;
        }

        @Override
        public void run() {
            try {
                box.dealHeartBeatPing(ping, oldRole, oldStamp);
            } finally {
                doing.compareAndSet(true, false);
            }
        }

    }

    private RaftMessage waitHeartBeat() {
        //long wait = voteRandomMillis();
        //LocalSleep.park(wait, TimeUnit.MILLISECONDS);
        try {
            RaftMessage ping = heartNotify.poll(5600, TimeUnit.MILLISECONDS);
            return ping;
        } catch (InterruptedException e) {
            return null;
        }
    }

    private long voteRandomMillis() {
        long base = Meta.min_elc_time.millis();
        long dura = Meta.max_elc_time.millis() - base;
        long sleep = base + random.nextLong(dura);
        return sleep;
    }

    private long heartRandomMillis() {
        long base = Meta.min_heart_time.millis();
        long dura = Meta.max_heart_time.millis() - base;
        long sleep = base + random.nextLong(dura);
        return sleep;
    }

    private long logRandomMillis() {
        long base = Meta.min_elc_time.millis();
        long dura = Meta.max_elc_time.millis() - base;
        long sleep = base + random.nextLong(dura);
        return sleep;
    }

    public boolean worldStopped() {
        if (start.get()) {
            return false;
        } else {
            return true;
        }
    }

    private final class SendHeartTask implements Runnable {

        private long lastTime = System.currentTimeMillis();

        public SendHeartTask() {

        }

        @Override
        public void run() {
            // 随机开始
            LocalSleep.park(voteRandomMillis(), TimeUnit.MILLISECONDS);
            for (; ; ) {
                if (worldStopped()) {
                    return;
                }

                LocalSleep.park(heartRandomMillis(), TimeUnit.MILLISECONDS);

                Role role = role();
                if (role.type() != Role.Name.LEADER) {
                    LocalSleep.safePoint();
                    continue;
                }

                final Lock lock = mainLock;
                lock.lock();
                try {
                    // 此处可以 debug 测试动态 主节点切换
                    role = role();
                    long now = System.currentTimeMillis();
                    if (role.type() != Role.Name.LEADER) {
                        LocalSleep.safePoint();
                        //log.info("auto send heart beat, i am leader , name = " + name);
                        System.out.println("auto send heart beat, i am leader , name = " + name);
                        lastTime = now;
                        continue;
                    } else {
                        //log.info("auto send heart beat, i am leader , name = " + name);
                        System.out.println("auto send heart beat, i am leader , name = " + name);
                        if ((now - lastTime) < Meta.min_heart_time.millis()) {
                            LocalSleep.safePoint();
                            lastTime = now;
                            continue;
                        } else {
                            heartBeatOthers(role);
                            lastTime = System.currentTimeMillis();
                        }
                    }
                } finally {
                    lock.unlock();
                }
            }
        }

    }

    public void heartBeatOthers(Role role) {
        List<NodeName> others = box.group().exceptSelf();
        for (NodeName remote : others) {
            HeartBeatPing ping = new HeartBeatPing();
            ping.setLeader(name());
            ping.setTerm(role.getTerm());
            ping.setLeaderCommit(0);
            ping.setPrevLogIndex(0);
            ping.setPrevLogTerm(0);
            String msgId = MsgID.getInstance().msgId();
            RaftMessage message = new RaftMessage(msgId, ping, name, remote);
            box.sendMessage(message);
            if (worldStopped()) {
                return;
            }
        }
    }

}
